package cn.monkey.transport.mq.rabbitmq;

import cn.money.transport.mq.MqInboundConfig;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import java.util.function.Supplier;

public class SimpleRabbitMqInbound implements RabbitMqInbound {

    private final RabbitMqInboundConfig inboundConfig;
    protected final Flux<byte[]> receiver;
    protected final Collection<Channel> channels;
    protected final Supplier<Mono<Void>> closeAction;

    public SimpleRabbitMqInbound(Collection<Channel> channels,
                                 RabbitMqInboundConfig inboundConfig,
                                 Supplier<Mono<Void>> closeAction) {
        this.inboundConfig = inboundConfig;
        this.channels = channels;
        this.closeAction = closeAction;
        this.receiver = Flux.fromArray(inboundConfig.getTopics())
                .flatMap(topic -> Flux.create(sink -> {
                    CompletableFuture.runAsync(() -> {
                        for (Channel channel : channels) {
                            try {
                                channel.basicConsume(topic, false, (consumerTag, message) -> {
                                    sink.next(message.getBody());
                                }, consumerTag -> {
                                });
                            } catch (IOException e) {
                                sink.error(e);
                            }
                        }
                    });
                }));
    }

    @Override
    public MqInboundConfig inboundConfig() {
        return inboundConfig;
    }

    @Override
    public Mono<Void> close() {
        return this.closeAction.get();
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.map(bytes -> this.inboundConfig.getDecoder().decode(bytes))
                .filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }
}
